package es.uam.eps.maprednat;

import java.io.*;
import java.sql.Time;
import java.text.SimpleDateFormat;
import java.util.Date;

import org.apache.hadoop.conf.*;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.*;
import org.apache.hadoop.mapreduce.lib.output.*;
import org.apache.hadoop.util.*;
import org.apache.commons.logging.*;

import org.gcn.plinguacore.parser.*;
import org.gcn.plinguacore.parser.input.*;
import org.gcn.plinguacore.simulator.*;
import org.gcn.plinguacore.util.*;
import org.gcn.plinguacore.util.psystem.*;
import org.gcn.plinguacore.util.psystem.cellLike.membrane.CellLikeMembrane;
import org.gcn.plinguacore.util.psystem.cellLike.membrane.CellLikeSkinMembrane;
import org.gcn.plinguacore.util.psystem.membrane.Membrane;

import sun.management.MemoryUsageCompositeData;


public class MapRedNat extends Configured implements Tool {
	static enum Counters { MEMBRANES, STOPPED };

	public static class MapClass extends
			Mapper<LongWritable, Text, Text, MembraneJoin> {

		static String pSystemTemplateStart = "@model<transition>\ndef main()\n{\n";
		static String pSystemTemplateEnd = "\n}";

		Psystem gps;
		Log log;
		
		public void setup(Context context) {
			log = LogFactory.getLog(Job.class);

			try {
				// Open a P-Lingua file

				FileInputStream stream = new FileInputStream(DistributedCache.getCacheFiles(context.getConfiguration())[0].getPath());
				try {
					// Create a new InputParser for P-Lingua format
					AbstractParserFactory ipf = new InputParserFactory();
					InputParser iparser = (InputParser) ipf.createParser("P-Lingua");
					iparser.setVerbosityLevel(5);
					// Parse the file
					gps = iparser.parse(stream);
				} catch (PlinguaCoreException ex) {
					ex.printStackTrace();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}

		public void map(LongWritable key, Text value, Context context)
				throws IOException {

			Log log = LogFactory.getLog(Job.class);
			log.info("Va a parsear: " + value.toString());
			String[] parts = value.toString().split(":");
			String contents = "";
			if (parts.length>3) {
				contents = "@ms("+parts[1]+")="+parts[3]+";";
			}
			String mem = pSystemTemplateStart+"@mu="+parts[2]+";"+contents + pSystemTemplateEnd;
			log.info("Membrana: " + mem);
			
			try {
				// Create a new InputParser for P-Lingua format
				AbstractParserFactory ipf = new InputParserFactory();
				InputParser iparser = (InputParser) ipf.createParser("P-Lingua");
				iparser.setVerbosityLevel(5);
				
				Psystem ps = iparser.parse(new StringBufferInputStream(mem));
				ps.getRules().addAll(gps.getRules());
				
				log.info("ps:"+ps.toString());
				
				// Create a simulator for active membranes
				ISimulator sim = ps.createSimulator(false, false);
				sim.setTimed(true);
				sim.setVerbosity(1);
				boolean nonStop = sim.step();

				CellLikeSkinMembrane sm = (CellLikeSkinMembrane)sim.getCurrentConfig().getMembraneStructure();
				CellLikeMembrane me;
				log.info(sm.toString());
				if (sm.getLabel().equals(parts[1]))
				{
					me = sm;	// no tiene membrana padre, es la piel
				} 
				else 
				{
					me = null;
					
					// busca la membrana actual dentro de la membrana padre. puede no estar por haberse disuelto
					for (CellLikeMembrane m: sm.getChildMembranes())
					{
						if (m.getLabel().equals(parts[1])){
							me = m;
							break;
						}
					}
					
					// la membrana padre recibe simbolos
					if (!sm.getMultiSet().isEmpty())
					{
						context.write(new Text(sm.getLabel()), new MembraneJoin(null, null, sm.getMultiSet().toString(), me==null?"[]'"+parts[1]+","+parts[0]:null));
					}
				}
				
				if (me!=null) {
					// mi estructura actualizada
					String membranes="", symbols = null;
					for (Membrane m: me.getChildMembranes())
					{
						if (membranes.length()>0) membranes+=" ";
						membranes += "[]'"+m.getLabel()+","+m.getLabelObj().getEnvironmentID();
					}
					membranes = "[" + membranes + "]'" + me.getLabel();
					if (!me.equals(sm)) membranes = "[" + membranes + "]'" + sm.getLabel();
					
					for (String s: me.getMultiSet())
					{
						if (s.contains("[")) continue;
						if (symbols==null) symbols = ""; else symbols += ", ";
						symbols += s;
					}
					
					context.write(new Text(parts[0]), new MembraneJoin(parts[1], membranes, symbols, null));
					
				}
				
				// envia a simbolos hijos, incluso aunque se haya disuelto 
				CellLikeMembrane pch = me;
				if (me==null) pch = sm;
				for (CellLikeMembrane ch : pch.getChildMembranes())
				{
					if (ch.getMultiSet().isEmpty()) continue;
					
					String id = ch.getLabelObj().getEnvironmentID();
					context.write(new Text(id), new MembraneJoin(null, null, ch.getMultiSet().toString(), null));
				}

				context.getCounter(Counters.MEMBRANES).increment(1);
				if (!nonStop) context.getCounter(Counters.STOPPED).increment(1);
			} catch (PlinguaCoreException e) {
				log.info(e.getStackTrace());
			} catch (InterruptedException e) {
				log.info(e.getStackTrace());
			}
		}
	}

	public static class ReduceClass extends
			Reducer<Text, MembraneJoin, Text, MembraneJoin> {

		public void reduce(Text key, Iterable<MembraneJoin> values, Context context)
				throws IOException, InterruptedException {

			Log log = LogFactory.getLog(Job.class);
			log.info("Reduce!");
			MembraneJoin join = null;
			for (MembraneJoin m: values)
			{
				log.info("raw-->"+m.toString());
				if (join==null) 
					join = new MembraneJoin(m);
				else
					join.join(m);
			}

			log.info("join-->"+join.toString());
			context.write(key, join);
		}
	}

	static int printUsage() {
		System.out
				.println("maprednat [-m <maps>] [-r <reduces>] <input> <output>");
		ToolRunner.printGenericCommandUsage(System.out);
		return -1;
	}

	/**
	 * The main driver for word count map/reduce program. Invoke this method to
	 * submit the map/reduce job.
	 * 
	 * @throws IOException
	 *             When there is communication problems with the job tracker.
	 */
	public int run(String[] args) throws Exception {
		
		// Make sure there are exactly 1 parameters left.
		if (args.length != 1) {
			System.out.println("ERROR: Wrong number of parameters: " + args.length + " instead of 1.");
			return printUsage();
		}

		Counter mems, memstops;
		String input = "input";
		SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd.HHmmss");
		String date = sdf.format(new Date());
		int step = 0;
		String output = date+"/"+step;
		
		boolean success;
		do {
			Job conf = new Job(getConf());
			conf.setJarByClass(MapRedNat.class);
			conf.setJobName("maprednat");
	
			conf.setMapperClass(MapClass.class);
			conf.setCombinerClass(ReduceClass.class);
			conf.setReducerClass(ReduceClass.class);
	
			conf.setOutputKeyClass(Text.class);
			conf.setOutputValueClass(MembraneJoin.class);
			conf.setInputFormatClass(TextInputFormat.class);
			conf.setOutputFormatClass(TextOutputFormat.class);
			conf.getConfiguration().set("mapreduce.output.textoutputformat.separator", ":");
	
			DistributedCache.addCacheFile(new Path(args[0]).toUri(), conf.getConfiguration());
			
			FileInputFormat.setInputPaths(conf, input);
			FileOutputFormat.setOutputPath(conf, new Path(output));
			success = conf.waitForCompletion(true);
			input = output;
			output = date+"/"+(++step);
			mems = conf.getCounters().findCounter(Counters.MEMBRANES);
			memstops = conf.getCounters().findCounter(Counters.STOPPED);
		} while (success && (mems.getValue() > memstops.getValue()));
		
		return success ? 0 : 1;
	}

	public static void main(String[] args) throws Exception { 
		int res = ToolRunner.run(new MapRedNat(), args); 
		System.exit(res); 

		try { // Open a P-Lingua file
			FileInputStream stream = new FileInputStream("test.pli");
			FileInputStream stream2 = new FileInputStream("test2.pli");
			try { // Create a new InputParser for P-Lingua format
				AbstractParserFactory pf = new InputParserFactory();
				InputParser parser = (InputParser) pf.createParser("P-Lingua");
				parser.setVerbosityLevel(5); // Parse the file
				Psystem ps = parser.parse(stream); // Create a simulator for
													// active membranes
				
				
				//Psystem tps = parser.parse(new StringBufferInputStream("@model<transition>\ndef main()\n{\n@mu=[[[]'3 []'4 []'5]'2]'1,101;\n@ms(3) = a,f;\n}")); // Create a simulator for
				// active membranes
				//System.out.println(ps.getMembraneStructure().toString());
				ISimulator sim = ps.createSimulator(false, false);
				//ps.setMembraneStructure(tps.getMembraneStructure());
				//System.out.println(ps.getMembraneStructure().toString());
				sim.setTimed(true);
				sim.setVerbosity(1); // Run until a halt state sim.run();
				sim.run();

				for (Membrane m: sim.getCurrentConfig().getMembraneStructure().getAllMembranes())
				{
					System.out.println(m.getLabel()+":"+m.getMultiSet());
				}
			} catch (PlinguaCoreException ex) {
				ex.printStackTrace();
			}
		} catch (FileNotFoundException e) {
			e.printStackTrace();
		}
	}
}
